﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using XcoAppSpaces.Core.Ports;
using XcoAppSpaces.Core;
using System.Threading;
using NUnit.Framework;

namespace test.XcoAppSpaces.Core.Error_Reproductions
{
    [TestFixture]
    public class testFilteredSubscribe
    {

        public static int ITEMCOUNT = 5000;
        private int CLIENTCOUNT = 5;

     /*   public static void Main(string[] args)
        {
            testFilteredSubscribe tfs = new testFilteredSubscribe();
            tfs.NormalSubscribe();
        }*/

        private ManualResetEvent waitForShutdown;
        private ManualResetEvent spaceStarted;
        private bool result = false;
        
        private double calcAllItems;
        private double calcDuration;
        
        private static double throughputItemsInSec = 0;
        private int clientShutdownCount = 0;
        private int clientKillCount = 0;

		//[Test]
		//public void NormalTest_5x500()
		//{
		//	ITEMCOUNT = 500;
		//	CLIENTCOUNT = 5;
		//	clientShutdownCount = 0;
		//	//clientKillCount = 0;
		//	TestSubscribe();
		//	Assert.IsTrue(result);
		//}

        [Test]
        public void NormalTest_3x30()
        {
            ITEMCOUNT = 30;
            CLIENTCOUNT = 3;
            clientShutdownCount = 0;
            //clientKillCount = 0;
            TestSubscribe();
            Assert.IsTrue(result);
        }

		/*[Test]
		public void NormalTest_6x2000()
		{
			ITEMCOUNT = 2000;
			CLIENTCOUNT = 6;
			clientShutdownCount = 0;
			clientKillCount = 0;
			TestSubscribe();
			Assert.IsTrue(result);
		}
		*/

		//[Test]
		//public void Shutdown_1_Test_5x500()
		//{
		//    ITEMCOUNT = 500;
		//    CLIENTCOUNT = 5;
		//    clientShutdownCount = 1;
		//    clientKillCount = 0;
		//    TestSubscribe();
		//    Assert.IsTrue(result);
		//}
		
		//[Test]
		//public void Kill_Test_5x500()
		//{
		//    ITEMCOUNT = 500;
		//    CLIENTCOUNT = 5;
		//    clientShutdownCount = 0;
		//    clientKillCount = 2;
		//    TestSubscribe();
		//    Assert.IsTrue(result);
		//}

       
        private void TestSubscribe()
        {
            result = false;

            waitForShutdown = new ManualResetEvent(false);
            spaceStarted = new ManualResetEvent(false);

            Thread tServer = new Thread(new ThreadStart(StartSpace));
            tServer.Start();

            spaceStarted.WaitOne();

            List<CountTest> ctList = new List<CountTest>();
            for (int i = 0; i < CLIENTCOUNT; i++)
            {
                CountTest ct = new CountTest(i);
                ctList.Add(ct);
                ct.Start();
                Thread.Sleep(250);
            }

            WaitForWorkers(ctList);

            DateTime startTime = DateTime.Now;
            foreach (CountTest ct in ctList)
            {
                ct.StartSend();
            }
            Thread.Sleep(1000);
            while (clientShutdownCount > 0)
            {
                if (ctList.Count <= 0)
                {//not possible
                    break;
                }
                ctList[0].Shutdown();
                ctList.RemoveAt(0);
                clientShutdownCount--;
            }
			while (clientKillCount > 0)
			{
				if (ctList.Count <= 0)
				{//not possible
					break;
				}
				ctList[0].Kill();
				ctList.RemoveAt(0);
				clientKillCount--;
			}
            CLIENTCOUNT = ctList.Count;

            WaitForFinish(ctList);
            int allItems = 0;
            foreach (CountTest ct in ctList)
            {
                allItems += ct.AllNumbers + ct.EvenNumbers + ct.OddNumbers;
            }
            Console.WriteLine("All filtered: " + (allItems) + " items in " + DateTime.Now.Subtract(startTime).TotalSeconds + " sec.");


            foreach (CountTest ct in ctList)
            {
                ct.Shutdown();
            }
            result = true;
            foreach (CountTest ct in ctList)
            {
                if (result)
                {
                    result = CheckResults(ct);
                }
            }
            waitForShutdown.Set();

            double durationItemsInSec = calcAllItems / calcDuration;
            if (throughputItemsInSec != 0)
            {
                 Console.WriteLine("Diff : " + (throughputItemsInSec - durationItemsInSec));
            }
            throughputItemsInSec = durationItemsInSec;
            Console.WriteLine("All items per sec processed: " + durationItemsInSec);

			Thread.Sleep(2000);
        }

        private void WaitForWorkers(List<CountTest> ctList)
        {
            int maxWait = 30 * 1000;
            int sleepInterval = 50;
            while (true)
            {
                Thread.Sleep(sleepInterval);
                int readyCount = 0;
                foreach (CountTest ct in ctList)
                {
                    if (ct.CurrentState == state.workerloaded)
                    {
                        readyCount++;
                    }
                }
                if (readyCount == ctList.Count)
                {//all ready!
                    break;
                }
                maxWait -= sleepInterval;
                if (maxWait < 0)
                {
                    break;
                }
            }
        }
        private void WaitForFinish(List<CountTest> ctList)
        {
            List<CountTest> workerList = new List<CountTest>();
            foreach (CountTest ct in ctList)
            {
                workerList.Add(ct);
            }
            int maxWait = 60 * 1000;
            int sleepInterval = 100;
            while (true)
            {
                Thread.Sleep(sleepInterval);

                for (int i = workerList.Count - 1; i >= 0; i--)
                {
                    CountTest ct = ctList[i];
                    if (CheckClient(ct))
                    {
                        
                        int receivedItems = ct.AllNumbers + ct.EvenNumbers + ct.OddNumbers;
                        int processedItems = ct.AllNumbers * 3;
                        double duration = DateTime.Now.Subtract(ct.startDate).TotalSeconds;
                        double durationItemsInSec = processedItems / duration;
                        calcAllItems += processedItems;
                        calcDuration += duration;

                        Console.WriteLine("Finished filtered: Received " + receivedItems+ " Processed: " +processedItems+" in " + duration + " sec items per sec processed: "+durationItemsInSec );




                        workerList.RemoveAt(i);
                    }
                }
                if (workerList.Count == 0)
                {//all ready!
                    break;
                }
                maxWait -= sleepInterval;
                if (maxWait < 0)
                {
                    break;
                }
            }
        }

        private bool CheckClient(CountTest ct)
        {
            return ct.AllNumbers >= ITEMCOUNT * CLIENTCOUNT;
        }

        private bool CheckResults(CountTest ct)
        {
            int allItemsCount = ITEMCOUNT * CLIENTCOUNT;
            bool result = ct.AllNumbers >= allItemsCount;
            if (result)
            {
                result = ct.OddNumbers >= allItemsCount / 2;
            }
            if (result)
            {
                result = ct.EvenNumbers >= allItemsCount / 2;
            }
            if (!result)
			{
				Console.WriteLine("Error Data: expected {0}, all {1}, odd {2}, even {3}", allItemsCount, ct.AllNumbers, ct.OddNumbers, ct.EvenNumbers);
			}
            return result;

        }
        private void StartSpace()
        {
			using (XcoAppSpace space = new XcoAppSpace("tcp.port=9010"))
            {
                space.RunWorker<TestWorker, PublishWorker>("remoteworker");
                Console.WriteLine("Worker started.");
                //Thread.Sleep(1000);
                spaceStarted.Set();
                //wait for stop
                waitForShutdown.WaitOne(5 * 60 * 1000);
				Thread.Sleep(1000);
                Console.WriteLine("Worker finished.");
            }
        }

    }

    public enum state
    {
        init,
        starting,
        workerloaded,
        running
    }
    public class CountTest
    {

        public DateTime startDate { get; private set; }

        public int AllNumbers { get { return allNumbers; } }
        public int OddNumbers { get { return oddNumbers; } }
        public int EvenNumbers { get { return evenNumbers; } }
        public state CurrentState { get; private set; }

        private int allNumbers;
        private int oddNumbers;
        private int evenNumbers;

        private ManualResetEvent waitStartWork; 
        private ManualResetEvent waitForShutdown;

        private bool unregister = true;
        public void Start()
        {
            CurrentState = state.init;
            Thread tServer = new Thread(new ThreadStart(StartClient));
            tServer.Start();
        }
        public void StartSend()
        {
            waitStartWork.Set();
        }
        public void Shutdown()
        {
            unregister = true;
            waitForShutdown.Set();
        }
        public void Kill()
        {
            unregister = false;
            waitForShutdown.Set();
        }
        public int ix { get; private set; }

        public CountTest(int ix)
        {
            this.ix = ix;
            waitStartWork = new ManualResetEvent(false);
            waitForShutdown = new ManualResetEvent(false);
        }
        private void StartClient()
        {
            int currentPort = 9500 + ix;

            try
            {
                allNumbers = 0;
                oddNumbers = 0;
                evenNumbers = 0;
                
                using (XcoAppSpace space = new XcoAppSpace("tcp.port=" + currentPort))
                {
                    CurrentState = state.starting;
                    TestWorker worker = space.ConnectWorker<TestWorker>("localhost:9010/remoteworker");

                    Port<TestWorker.Notification> subAllNumbers = new Port<TestWorker.Notification>();
                    Port<TestWorker.Notification> subOddNumbers = new Port<TestWorker.Notification>();
                    Port<TestWorker.Notification> subEvenNumbers = new Port<TestWorker.Notification>();
                    Port<Exception> exceptionPort = new Port<Exception>();
					Causality c = new Causality(exceptionPort);
					CausalityContext.Add(c);


	                subAllNumbers.RegisterReceiver(n => Interlocked.Increment(ref allNumbers));
	                subOddNumbers.RegisterReceiver(n => Interlocked.Increment(ref oddNumbers));
                    subEvenNumbers.RegisterReceiver(n => Interlocked.Increment(ref evenNumbers));
                    exceptionPort.RegisterReceiver(ex => Console.WriteLine("ERROR: " + ex));

                    // Subscribe to a message type with a server side filter. Pass in a
                    // lambda expression to be executed by the worker (!) for each notification. 
                    // Check if the notification should be sent at all to the subscription port.
                    worker.Post(new Subscribe<TestWorker.Notification>(subAllNumbers));

                    worker.Post(new Subscribe<TestWorker.Notification>(
                                    subOddNumbers,
                                    n => n.Number % 2 == 1
                                    ));
                    worker.Post(new Subscribe<TestWorker.Notification>(
                                    subEvenNumbers,
									n => n.Number % 2 == 0
                                    ));
                    
                    Console.WriteLine("Client[" + currentPort + "] is waiting ");
                    Thread.Sleep(250);
                    CurrentState = state.workerloaded;
                    waitStartWork.WaitOne(15 * 1000); //sync wait for other clients

                    Console.WriteLine("Client[" + currentPort + "] starts ");
                    startDate = DateTime.Now;
                    CurrentState = state.running;
                    for (int i = 1; i <= testFilteredSubscribe.ITEMCOUNT; i++)
                    {
                        worker.Post(i);
                    }
                    Console.WriteLine("Post[" + currentPort + "]: " + testFilteredSubscribe.ITEMCOUNT + " in " + DateTime.Now.Subtract(startDate).TotalSeconds + " sec.");
                    startDate = DateTime.Now;

                    waitForShutdown.WaitOne(60 * 1000);
                    if (unregister)
                    {
                        worker.Post(new Unsubscribe<TestWorker.Notification>(subAllNumbers));
                        worker.Post(new Unsubscribe<TestWorker.Notification>(subEvenNumbers));
                        worker.Post(new Unsubscribe<TestWorker.Notification>(subOddNumbers));
                        Console.WriteLine("Client[" + currentPort + "]  unsubscribed ");
						Thread.Sleep(1000);
                    }
					CausalityContext.RemoveCurrent();

                    Console.WriteLine("Client[" + currentPort + "] finished.");
                }
            }
            catch (Exception err)
            {
                Console.Error.WriteLine("Client[" + currentPort + "]: " + err.Message);
            }
        }
    }


    class TestWorker : PortSet<int, Subscribe<TestWorker.Notification>, Unsubscribe<TestWorker.Notification>>
    {
        [Serializable]
        public class Notification
        {
            public int Number;
        }
    }

    class PublishWorker : TestWorker
    {
        private readonly XcoPublisher<Notification> subscriptions
                    = new XcoPublisher<Notification>();


        [XcoConcurrent]
        void ProcessInt(int n)
        {
            subscriptions.Publish(new Notification { Number = n });
        }
    }
}